#include "EventLoop.h"
#include "Logger.h"
#include "Poller.h"
#include "Channel.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <memory>

// 防止一个线程创建多个Eventloop
__thread EventLoop *t_loopInThisThread = nullptr;

// 定义默认的Poller IO复用接口的超时时间
const int kPollTimeMs = 10000;

// 创建wakeupfd，用来notify唤醒subReactor处理新来的channel
int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_FATAL("eventfd error:%d \n", errno);
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false)
    , quit_(false)
    , callingPendingFunctors_(false)
    , threadId_(CurrentThread::tid())
    , poller_(Poller::newDefaultPoller(this))
    , wakeupFd_(createEventfd())
    , wakeupChannel_(new Channel(this, wakeupFd_))
{
    LOG_DEBUG("EventLoop created %p in thread %d \n", this, threadId_);
    if (t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this therad %d \n", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    // 设置wakeupfd的事件类型以及发生事件后的回调操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 每一个eventloop都将监听wakeupchannel的EPOLLIN读事件了
    wakeupChannel_->enableReading();
}
EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

// 开启事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;
    LOG_INFO("EventLoop %p start looping. \n", this);
    while (!quit_)
    {
        activeChannels_.clear();
        // EventLoop可以想象成poll poll监听两类fd，一种是client通信的fd，一种是唤醒subLoop的wakeupFd
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
        for (Channel *channel : activeChannels_)
        {
            // Poller监听哪些channel发生事件了，然后上报给EventLoop，通知channel处理事件
            channel->handleEvent(pollReturnTime_);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        /**
         * eg：
         * IO线程（主线程mainLoop负责accept fd监听新连接，打包成一个channel），如果是已经发生事件的channel，就要将channel注册分发给工作线程subLoop
         * 但是mainLoop不能直接操作subLoop的Poller，只能委托subLoop自己处理
         * 主线程调用subLoop的queueInLoop()方法，将"注册Channel"的回调加入pendingFunctors_队列
         * 主线程再调用wakeup()唤醒subLoop（subLoop在poll阻塞）
         * subLoop线程被唤醒之后，在doPendingFunctors()中，执行之前mainLoop注册的"注册Channel"回调操作
         * 这时候操作的是自己的Poller，不会有线程安全问题
         */
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping. \n", this);
    looping_ = false;
}

// 退出事件循环
/**
 * 退出eventloop有两种情况 1.loop在自己的线程中调用quit 2.在非loop的线程中，调用loop的quit
 *                              mainLoop
 *
 *                              waupFd
 *                                                                          no =============== 生产者-消费者的线程安全队列 mainLoop向队列写channel subLoop从队列取channel
 *
 *              subLoop1        subLoop2         subLoop3
 * 比如：subLoop1在subLoop2中调用quit，想退出subLoop2；或者在mainLoop中调用quit，想退出mainLoop都需要wakeupFd先唤醒另一个线程
 */
void EventLoop::quit()
{
    quit_ = true;
    // EventLoop的quit（）可能被其他线程调用，比如mainLoop调用subLoop的quit（）退出subLoop，需要先唤醒subLoop
    if (!isInLoopThread())
    {
        wakeup();
    }
}

// 在当前loop中执行回调cb
void EventLoop::runInloop(Functor cb)
{
    if (isInLoopThread()) // 在当前的loop线程中，执行cb
    {
        cb();
    }
    else // 在非当前的loop线程中执行cb，就需要唤醒loop所在的线程，执行cb。eg：subLoop1需要执行subLoop2的runInLoop，subLoop1就要唤醒subLoop2
    {
        queueInLoop(cb);
    }
}
// 把cb放入队列中，唤醒loop所在的线程，执行cb
void EventLoop::queueInLoop(Functor cb)
{
    // 加锁允许其他线程安全的提交需要执行的操作（回调函数）
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    // 唤醒相应的，需要执行上面回调操作的loop的线程
    /**
     * || callingPendingFunctors_的意思是：当前loop正在执行回调，但是loop自己又给自己添加了新的回调，也需要wakeup（）唤醒，
     * 保证当前loop也能处理这个新的回调。因为执行dopendingFunctors（）开始时已经通过swap将当时所有的回调取走了
     * 因此需要wakeup（）唤醒自己，让EventLoop执行完当前回调之后，不进入poll阻塞，直接再次执行dopendingFunctors（）处理新添加的回调，避免新回调延迟
     */
    if (!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup(); // 唤醒loop所在的线程
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR("EventLoop::handleRead() reads %d bytes instad of 8", n);
    }
}

// 用来唤醒loop所在的线程的 向wakeupFd_上写一个数据，wakeupChannel就发生了读事件，当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1; 
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8 \n", n);
    }
}

// EventLoop的方法 =》 Poller的方法
void EventLoop::updateChannel(Channel *channel)
{
    poller_->updateChannel(channel);
}
void EventLoop::removeChannel(Channel *channel)
{
    poller_->removeChannel(channel);
}
bool EventLoop::hasChannel(Channel *channel)
{
    return poller_->hasChannel(channel);
}

// 执行回调
void EventLoop::doPendingFunctors()
{
    /**
     * 定义一个新的functors容器，和存储回调操作的pendingFunctors_进行交换。为什么不直接加锁遍历pendingFuncotrs_?
     * 锁竞争激烈。执行回调的过程可能很慢，锁被长时间占有，其他线程调用queueInloop（）添加新的回调时会阻塞在加锁步骤，导致新回调无法及时添加
     * 定义一个新functor，和pendingFunctors_交换目的是：
     * 通过swap（）瞬间完成回调的转移，将持有锁的时间压缩到最短
     * 用新容器存储待执行的回调，其他线程可以继续添加回调到pendingFuncotrs_，使添加回调和执行回调可以并发进行
     * 平衡了线程安全和执行效率，避免回调处理成为性能瓶颈
     */
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;
    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for (const Functor &functor : functors)
    {
        functor(); // 执行当前loop需要执行的回调操作
    }

    callingPendingFunctors_ = false;
}